import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 读写调度
 */
public class DispatchContext {
    private AtomicBoolean isClosed = new AtomicBoolean(false);
    // 是否处于某个过程
    private AtomicBoolean inRegInput = new AtomicBoolean(false);
    private AtomicBoolean inRegOutput = new AtomicBoolean(false);
    private Selector readSelector;
    private Selector writeSelector;
    private ServerAcceptor serverAcceptor;
    private SocketCloseInterface socketCloseInterface;

    public DispatchContext(ServerAcceptor serverAcceptor) {
        this.serverAcceptor = serverAcceptor;
    }


    public void dispatchStart() {
        try {
            this.socketCloseInterface = serverAcceptor.getSocketCloseClass();
            readSelector = Selector.open();
            writeSelector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
        startRead();
        startWrite();
    }

    public void exit() {
        if (isClosed.compareAndSet(false, true)) {
            CloseUtils.close(readSelector, writeSelector);
            System.out.println("DispatchContext 关闭");
        }
    }

    /**
     * @param channel     客户端通道
     * @param selector    选择器
     * @param registerOps 感兴趣的状态
     * @param locker      锁
     * @return
     */
    private SelectionKey registerSelection(SocketChannel channel, Selector selector,
                                           int registerOps, AtomicBoolean locker) {

        //noinspection SynchronizationOnLocalVariableOrMethodParameter
        synchronized (locker) {
            // 设置锁定状态
            locker.set(true);

            try {
                // 唤醒当前的selector，让selector不处于select()状态
                selector.wakeup();

                SelectionKey key = null;
                if (channel.isRegistered()) {
                    // 查询是否已经注册过
                    key = channel.keyFor(selector);
                    if (key != null) {
                        key.interestOps(key.readyOps() | registerOps);
                    }
                }

                if (key == null) {
                    // 注册selector得到Key
                    key = channel.register(selector, registerOps);
                    // 注册回调
                }

                return key;
            } catch (ClosedChannelException
                    | CancelledKeyException
                    | ClosedSelectorException e) {
                return null;
            } finally {
                // 解除锁定状态
                locker.set(false);
                try {
                    // 通知
                    locker.notify();
                } catch (Exception ignored) {
                }
            }
        }
    }

    private void unRegisterSelection(SocketChannel channel, Selector selector,
                                     AtomicBoolean locker) {
        //noinspection SynchronizationOnLocalVariableOrMethodParameter
        synchronized (locker) {
            locker.set(true);
            selector.wakeup();
            try {
                if (channel.isRegistered()) {
                    SelectionKey key = channel.keyFor(selector);
                    if (key != null) {
                        // 取消监听的方法
                        key.cancel();
                    }
                }
            } finally {
                locker.set(false);
                try {
                    locker.notifyAll();
                } catch (Exception ignored) {
                }
            }
        }
    }

    public void registerConnectorHandler(ConnectorHandler connectorHandler, int ops) {

    }

    public void register(SocketChannel socketChannel, int ops) throws Exception {
        SelectionKey key;
        if (ops == SelectionKey.OP_READ) {
            key = registerSelection(socketChannel, readSelector, SelectionKey.OP_READ, inRegInput);
        } else {
            key = registerSelection(socketChannel, writeSelector, SelectionKey.OP_WRITE, inRegOutput);
        }
        if (key == null) {
            throw new IOException("Register error: channel:" + socketChannel + " ops:" + ops);
        }
    }

    public void unregister(SocketChannel channel) {
        unRegisterSelection(channel, readSelector, inRegInput);
//        unRegisterSelection(channel, writeSelector, inRegOutput);
    }

    private void startRead() {
        Thread thread = new SelectThread("Clink IoSelectorProvider ReadSelector Thread",
                isClosed, inRegInput, readSelector,
                SelectionKey.OP_READ, this, socketCloseInterface, serverAcceptor);
        thread.start();
    }

    private void startWrite() {
        Thread thread = new SelectThread("Clink IoSelectorProvider WriteSelector Thread",
                isClosed, inRegOutput, writeSelector,
                SelectionKey.OP_WRITE, this, socketCloseInterface, serverAcceptor);
        thread.start();
    }


    /**
     * 读或写的线程
     */
    static class SelectThread extends Thread {
        private final AtomicBoolean isClosed;
        private final AtomicBoolean locker;
        private final Selector selector;
        private final int keyOps;
        private final DispatchContext dispatchContext;
        private SocketCloseInterface socketCloseInterface;
        private ServerAcceptor serverAcceptor;
        //        private ByteBuffer byteBuffer = ByteBuffer.allocate(512);
        private volatile HashMap<String, ConnectorHandler> handlerHashMap;

        private void waitSelection(final AtomicBoolean locker) {
            //noinspection SynchronizationOnLocalVariableOrMethodParameter
            synchronized (locker) {
                if (locker.get()) {
                    try {
                        locker.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        SelectThread(String name,
                     AtomicBoolean isClosed, AtomicBoolean locker,
                     Selector selector
                , int keyOps
                , DispatchContext dispatchContext
                , SocketCloseInterface socketCloseInterface
                , ServerAcceptor serverAcceptor) {
            super(name);
            this.isClosed = isClosed;
            this.locker = locker;
            this.selector = selector;
            this.keyOps = keyOps;
            this.dispatchContext = dispatchContext;
            this.socketCloseInterface = socketCloseInterface;
            this.serverAcceptor = serverAcceptor;
            this.setPriority(Thread.MAX_PRIORITY);
//            this.channelIndex = 1;
        }

        @Override
        public void run() {
            super.run();
            AtomicBoolean locker = this.locker;
            AtomicBoolean isClosed = this.isClosed;
            Selector selector = this.selector;
            int keyOps = this.keyOps;

            while (!isClosed.get()) try {
                if (selector.select() == 0) {
                    waitSelection(locker);
                    continue;
                } else if (locker.get()) {
                    waitSelection(locker);
                }

                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    if (selectionKey.isValid()) {
                        final SocketChannel channel = (SocketChannel) selectionKey.channel();
                        if (channel != null && channel.isOpen()) {
                            final String ipAndPort = channel.getRemoteAddress().toString();
                            if (selectionKey.isValid()) {
                                if (keyOps == SelectionKey.OP_READ) {
                                    ByteBuffer byteBuffer = ByteBuffer.allocate(IoContext.ReadBufferSize);
                                    try {
                                        long read = channel.read(byteBuffer);
                                        if (read > 0) {
                                            byteBuffer.flip();
                                            final ConnectorHandler connectorHandler = this.serverAcceptor.getSocketChannelMap().get(channel.getRemoteAddress().toString());
                                            if (connectorHandler != null) {
                                                if (byteBuffer.capacity() + 1 > byteBuffer.limit())
                                                    socketCloseInterface.onSocketMessageArrived(
                                                            connectorHandler
                                                            , byteBuffer);
                                            }
                                        }
                                        if (read == -1) {
                                            dispatchContext.unregister(channel);
                                            socketCloseInterface.SocketConnectorClose(ipAndPort);
                                            this.serverAcceptor.removeSocketList(ipAndPort);
                                            channel.close();
                                        }
                                    } catch (IOException e) {
                                        dispatchContext.unregister(channel);
                                        socketCloseInterface.SocketConnectorClose(ipAndPort);
                                        this.serverAcceptor.removeSocketList(ipAndPort);
                                        channel.close();
                                    } finally {
                                        byteBuffer.clear();
                                    }
                                }
                                if (keyOps == SelectionKey.OP_WRITE) {
                                    try {
                                        final ConnectorHandler connectorHandler = this.serverAcceptor.getSocketChannelMap().get(channel.getRemoteAddress().toString());
                                        if (connectorHandler != null) {
                                            if (connectorHandler.getClientType() == 0) {
                                                final Date lastWriteTime = connectorHandler.getLastWriteTime();
                                                final long l = new Date().getTime() - lastWriteTime.getTime() + IoContext.EquipmentChannelTime * IoContext.SensorCount;
                                                if (l > IoContext.EquipmentWriteTime) {
                                                    for (int i = 0; i < IoContext.SensorCount; i++) {
                                                        final byte[] command = IoContext.getCommand(
                                                                connectorHandler.getCommandAddress()
                                                                , i);
                                                        if (command != null && command.length > 0) {
                                                            ByteBuffer byteBuffer = ByteBuffer.allocate(command.length);
                                                            byteBuffer.put(command);
                                                            byteBuffer.flip();
                                                            connectorHandler.writeMessage(byteBuffer);
                                                            byteBuffer.clear();
                                                        }
                                                        if (i == 0)
                                                            connectorHandler.setLastWriteTime();
                                                        Thread.sleep(IoContext.EquipmentChannelTime);
                                                    }
                                                }
                                                Thread.sleep(IoContext.EquipmentWriteEvery);
                                            }
                                        }
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                }
                            }
                        }
                    }
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (ClosedSelectorException ignored) {
                break;
            }
        }


    }


}
